package com.heima.behavior.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.heima.behavior.entity.ApUnlikesBehavior;
import com.heima.behavior.mapper.ApUnlikesBehaviorMapper;
import com.heima.behavior.service.ApUnlikesBehaviorService;
import com.heima.common.constants.message.HotArticleConstants;
import com.heima.common.threadlocal.UserThreadLocalUtils;
import com.heima.common.util.JsonUtils;
import com.heima.model.article.dtos.ArticleBehaviorMsg;
import com.heima.model.article.dtos.UserBehaviorMsgDto;
import com.heima.model.behavior.dtos.UnLikesBehaviorDto;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.Map;

/**
 * <p>
 * APP不喜欢行为表 服务实现类
 * </p>
 *
 * @author HM
 * @since 2023-04-09
 */
@Slf4j
@Service
public class ApUnlikesBehaviorServiceImpl extends ServiceImpl<ApUnlikesBehaviorMapper, ApUnlikesBehavior> implements ApUnlikesBehaviorService {
    public static final String BEHAVIOR_KEY = "behavior:";
    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private KafkaProducer kafkaProducer;


    /**
     * 用户不喜欢行为
     */
    @Override
    public void unlikeBehavior(UnLikesBehaviorDto dto) {
        Integer userId = UserThreadLocalUtils.getUserId();
        String behaviorUserIdKey = BEHAVIOR_KEY + userId; // 存储 redis hash结构用的 大key
        /* 将用户行为数据写入redis */
        // 根据 type 判断用户行为（1 不喜欢，0 取消不喜欢）
        if (dto.getType().equals(UnLikesBehaviorDto.Type.UNLIKE.getCode())) {
            // 用户取消不喜欢
            behaviorDataAddRedis(dto, behaviorUserIdKey, false);
        }
        // 用户点击不喜欢
        behaviorDataAddRedis(dto, behaviorUserIdKey, true);
        /* 用户行为数据写入数据库 */
        // 查询数据库
        ApUnlikesBehavior unlikesBehaviorDB = getUnlikesBehaviorDB(dto, userId);
        // 数据库中存在当前用户关于当前文章的行为数据
        if (unlikesBehaviorDB != null) {
            unlikesBehaviorDB.setType(dto.getType());
            unlikesBehaviorDB.setUpdateTime(new Date());
            this.updateById(unlikesBehaviorDB);
        }
        // 数据库中不存在当前用户关于当前文章的行为数据
        saveUnlikesBehaviorDB(dto, userId);
    }

    /**
     * 将用户行为数据写入redis
     */
    private void behaviorDataAddRedis(UnLikesBehaviorDto dto, String behaviorUserIdKey, boolean isunlike) {
        UserBehaviorMsgDto redisMsg = getRedisMsgDto(dto, behaviorUserIdKey);
        redisMsg.setIsunlike(isunlike);
        redisTemplate.opsForHash().put(behaviorUserIdKey, dto.getArticleId(), redisMsg);
        sendMsgToKafka(dto);
    }

    /**
     * 数据库中不存在当前用户关于当前文章的行为数据，构建行为数据，写入数据库
     */
    private void saveUnlikesBehaviorDB(UnLikesBehaviorDto dto, Integer userId) {
        ApUnlikesBehavior unlikesBehavior = new ApUnlikesBehavior();
        unlikesBehavior.setArticleId(dto.getArticleId());
        unlikesBehavior.setApUserId(userId);
        unlikesBehavior.setEquipmentId(dto.getEquipmentId());
        unlikesBehavior.setType(dto.getType());
        unlikesBehavior.setCreatedTime(new Date());
        unlikesBehavior.setUpdateTime(new Date());
        this.save(unlikesBehavior);
    }

    /**
     * 设置查询条件查询数据库
     */
    private ApUnlikesBehavior getUnlikesBehaviorDB(UnLikesBehaviorDto dto, Integer userId) {
        LambdaQueryWrapper<ApUnlikesBehavior> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(ApUnlikesBehavior::getArticleId, dto.getArticleId());
        queryWrapper.eq(ApUnlikesBehavior::getApUserId, userId);
        return this.getOne(queryWrapper);
    }

    /**
     * 根据 用户ID 与 文章ID 组成的 key 获取到当前文章的用户行为数据
     *
     * @return 用于记录用户行为数据的类
     */
    private UserBehaviorMsgDto getRedisMsgDto(UnLikesBehaviorDto dto, String behaviorUserIdKey) {
        Map<Object, Object> userBehavior = redisTemplate
                .opsForHash()
                .entries(behaviorUserIdKey);
        //redis中未存在当前用户的行为数据
        if (userBehavior.isEmpty()) {
            return new UserBehaviorMsgDto();
        }
        // redis 中是否存在当前用户关于当前文章的行为数据
        UserBehaviorMsgDto msgDto = (UserBehaviorMsgDto) userBehavior.get(dto.getArticleId());
        return msgDto == null ? new UserBehaviorMsgDto() : msgDto;
    }

    /**
     * 将用户的行为数据发送到 Kafka
     */
    private void sendMsgToKafka(UnLikesBehaviorDto dto) {
        ArticleBehaviorMsg msg = new ArticleBehaviorMsg();
        msg.setArticleId(dto.getArticleId());
        msg.setNum(-dto.getType());
        msg.setType(ArticleBehaviorMsg.ArticleBehaviorType.VIEWS);
        ProducerRecord<String, String> record = new ProducerRecord<String, String>(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC, JsonUtils.toString(msg));
        kafkaProducer.send(record);
        log.info("发送 用户不喜欢 行为数据:{}", JsonUtils.toString(msg));
    }
}
